<?php
namespace app\controller;
//接收端 需要守护进程运行
use app\BaseController;
use PhpAmqpLib\Connection\AMQPStreamConnection;
class Consume  extends BaseController
{
    protected $consumerTag = 'consumer';
    protected $exchange = 'router';
    protected $queue = 'msgs';

    /*
    *关闭
    */
    function shutdown($channel, $connection)
    {
        $channel->close();
        $connection->close();
    }

    /*
    *回调处理信息
    */
    function process_message($message)
    {
        if ($message->body !== 'quit') {
            print_r($message->body);
        }
        //手动应答
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }

    /**
     * 启动 守护进程运行
     * 在public目录运行php index.php index/message_onsume/start &
     */
    public function start()
    {
        $host = '127.0.0.1';
        $port = 5672;
        $user = 'guest';
        $pwd = 'guest';
        $vhost = '/';
        $connection = new AMQPStreamConnection($host, $port, $user, $pwd, $vhost);
        $channel = $connection->channel();
        $channel->queue_declare($this->queue, false, true, false, false);
        $channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $channel->queue_bind($this->queue, $this->exchange);
        $channel->basic_consume($this->queue, $this->consumerTag, false, false, false, false, array($this, 'process_message'));

        register_shutdown_function(array($this, 'shutdown'), $channel, $connection);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }
}